package com.duobaoyu.dby.supply.biz.service.message.impl;

import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.duobaoyu.dby.channel.adapter.enums.biz.message.MessageStatusEnum;
import com.duobaoyu.dby.common.core.constant.CommonConstants;
import com.duobaoyu.dby.common.mybatis.support.ParamConvert;
import com.duobaoyu.dby.common.mybatis.util.EnumUtils;
import com.duobaoyu.dby.supply.biz.converter.ChannelMessageConverter;
import com.duobaoyu.dby.supply.biz.producer.SupplyChainMQProducer;
import com.duobaoyu.dby.supply.biz.service.message.MessageService;
import com.duobaoyu.dby.supply.biz.service.queue.core.DistributionQueueJobConsumerWork;
import com.duobaoyu.dby.supply.biz.strategy.AbstractChannelMessageAsyncHandler;
import com.duobaoyu.dby.supply.common.biz.bo.message.ChannelMessageBO;
import com.duobaoyu.dby.supply.common.biz.dto.message.*;
import com.duobaoyu.dby.supply.core.common.exception.CustomAssert;
import com.duobaoyu.dby.supply.core.common.util.DeflaterUtils;
import com.duobaoyu.dby.supply.core.dinger.util.DingDingUtils;
import com.duobaoyu.dby.supply.core.framework.util.FrameworkUtils;
import com.duobaoyu.dby.supply.core.rocketmq.bean.BaseMessageConsumer;
import com.duobaoyu.dby.supply.core.rocketmq.bean.BaseMqMessage;
import com.duobaoyu.dby.supply.core.rocketmq.constant.RocketMqConstant;
import com.duobaoyu.dby.supply.db.constant.SupplyMapperConstant;
import com.duobaoyu.dby.supply.db.entity.message.*;
import com.duobaoyu.dby.supply.db.helper.DynamicTableNameHelper;
import com.duobaoyu.dby.supply.db.repository.message.ChannelMessageRepository;
import com.duobaoyu.dby.supply.db.repository.message.DelayMessageRepository;
import com.duobaoyu.dby.supply.db.repository.message.MqMessageConsumerRepository;
import com.duobaoyu.dby.supply.db.repository.message.MqMessageRepository;
import com.duobaoyu.dby.supply.db.util.OperateAuthUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.SendStatus;
import org.springframework.context.annotation.Lazy;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.regex.Pattern;

/**
 * @author liyang
 * @version 1.0.0
 * @ClassName MessageServiceImpl
 * @createTime 15:24:00 2023/10/07
 */
@Slf4j
@Service
@RequiredArgsConstructor(onConstructor_ = {@Lazy})
public class MessageServiceImpl implements MessageService {

    private final MqMessageRepository mqMessageRepository;
    private final SupplyChainMQProducer supplyChainMqProducer;
    //    private final IdGeneratorSnowflake idGeneratorSnowflake;
    private final DelayMessageRepository delayMessageRepository;
    private final ChannelMessageConverter channelMessageConverter;
    private final ChannelMessageRepository channelMessageRepository;
    private final MqMessageConsumerRepository mqMessageConsumerRepository;
    private final DistributionQueueJobConsumerWork distributionQueueJobConsumerWork;


    @Override
    public boolean createBaseMessageConsumer(BaseMessageConsumer baseMessageConsumer) {
        if(log.isDebugEnabled()) {
            log.debug("供应链RocketMQ消费者消费消息存储: {}", JSON.toJSONString(baseMessageConsumer));
        }
        MqMessageConsumer mqMessageConsumer = MqMessageConsumer.builder()
                //.id(idGeneratorSnowflake.snowflakeId())
                .topic(baseMessageConsumer.getTopic())
                .tag(baseMessageConsumer.getTag())
                .uniqueKey(baseMessageConsumer.getUniqueKey())
                .status(baseMessageConsumer.getStatus())
                .msgId(baseMessageConsumer.getMsgId())
                .messageLinkId(baseMessageConsumer.getMessageLinkId())
                .messageOrigin(baseMessageConsumer.getMessageOrigin())
                .origin(baseMessageConsumer.getOrigin())
                .version(CommonConstants.ONE)
                .createBy(OperateAuthUtil.getAdministrator())
                .createTime(LocalDateTime.now())
                .updateBy(OperateAuthUtil.getAdministrator())
                .updateTime(LocalDateTime.now())
                .build();
        //return mqMessageConsumerRepository.save(mqMessageConsumer);
        return mqMessageConsumerRepository.save(mqMessageConsumer);
    }

    @Override
    public boolean createMqMessageStorage(BaseMqMessage baseMqMessage) {
        if(log.isDebugEnabled()) {
            log.debug("供应链RocketMQ生产消息存储: {}", JSON.toJSONString(baseMqMessage));
        }
        if (baseMqMessage.getIfSync() && Objects.equals(baseMqMessage.getStatus(), RocketMqConstant.FAIL)) {
            DingDingUtils.sendMarkdownMsg(JSON.toJSONString(baseMqMessage), "RocketMq消息同步发送失败");
        }
        MqMessage mqMessage = MqMessage.builder()
                //.id(idGeneratorSnowflake.snowflakeId())
                .topic(baseMqMessage.getTopic())
                .tag(baseMqMessage.getTag())
                .message(baseMqMessage.getMessage())
                .timeout(baseMqMessage.getTimeout())
                .hashKey(baseMqMessage.getHashKey())
                .status(baseMqMessage.getStatus())
                .msgId(baseMqMessage.getMsgId())
                .uniqueKey(baseMqMessage.getUniqueKey())
                .messageLinkId(baseMqMessage.getMessageLinkId())
                .origin(baseMqMessage.getOrigin())
                .version(CommonConstants.ONE)
                .createBy(OperateAuthUtil.getAdministrator())
                .createTime(LocalDateTime.now())
                .updateBy(OperateAuthUtil.getAdministrator())
                .updateTime(LocalDateTime.now())
                .build();
        return mqMessageRepository.save(mqMessage);
    }

    @Override
    public boolean updateMqMessageByUniqueKey(BaseMqMessage baseMqMessage) {
        if(log.isDebugEnabled()) {
            log.debug("供应链RocketMQ生产消息修改: {}", JSON.toJSONString(baseMqMessage));
        }
        if (Objects.equals(baseMqMessage.getStatus(), RocketMqConstant.FAIL)) {
            DingDingUtils.sendMarkdownMsg(JSON.toJSONString(baseMqMessage), "RocketMq消息异步发送失败");
        }
        /*return mqMessageRepository.update(Wrappers.<MqMessage>lambdaUpdate()
                .set(StringUtils.isNotBlank(baseMqMessage.getMsgId()),MqMessage::getMsgId,baseMqMessage.getMsgId())
                .set(MqMessage::getStatus,baseMqMessage.getStatus())
                .eq(MqMessage::getUniqueKey,baseMqMessage.getUniqueKey())
                .eq(StringUtils.isNotBlank(baseMqMessage.getTopic()),MqMessage::getTopic,baseMqMessage.getTopic())
                .eq(StringUtils.isNotBlank(baseMqMessage.getTag()),MqMessage::getTag,baseMqMessage.getTag())
                .eq(StringUtils.isNotBlank(baseMqMessage.getHashKey()),MqMessage::getHashKey,baseMqMessage.getHashKey()));*/
        Criteria criteria = new Criteria();
        criteria.and("unique_key").is(baseMqMessage.getUniqueKey());
        if (StringUtils.isNotBlank(baseMqMessage.getTopic())) {
            criteria.and("topic").is(baseMqMessage.getTopic());
        }
        if (StringUtils.isNotBlank(baseMqMessage.getTag())) {
            criteria.and("tag").is(baseMqMessage.getTag());
        }
        if (StringUtils.isNotBlank(baseMqMessage.getHashKey())) {
            criteria.and("hash_key").is(baseMqMessage.getHashKey());
        }
        Update update = new Update();
        update.inc("version", CommonConstants.ONE);
        update.set("msg_id", baseMqMessage.getMsgId());
        update.set("status", baseMqMessage.getStatus());
        update.set("update_time", LocalDateTime.now());
        update.set("update_by", OperateAuthUtil.getAdministrator());
        return mqMessageRepository.update(Query.query(criteria), update);
    }

    @Override
    public Page<MqMessage> mqMessagePage(MqMessageDTO mqMessage) {
        // 创建条件对象
        Criteria criteria = new Criteria();
        if(StringUtils.isNotBlank(mqMessage.getTopic())){
        	criteria.and("topic").is(mqMessage.getTopic());
		}
        if (StringUtils.isNotBlank(mqMessage.getTag())) {
            criteria.and("tag").is(mqMessage.getTag());
        }
        if (StringUtils.isNotBlank(mqMessage.getMessage())) {
            criteria.and("message").in(mqMessage.getMessage());
        }
        if (StringUtils.isNotBlank(mqMessage.getStatus())) {
            criteria.and("status").is(mqMessage.getStatus());
        }
        if (StringUtils.isNotBlank(mqMessage.getOrigin())) {
            criteria.and("origin").is(mqMessage.getOrigin());
        }
        if (StringUtils.isNotBlank(mqMessage.getMsgId())) {
            criteria.and("msg_id").is(mqMessage.getMsgId());
        }
        return mqMessageRepository.page(ParamConvert.getPage(mqMessage), Query.query(criteria));
    }

    @Override
    public List<MqMessage> mqMessageList(int skipCount, int limit, List<String> statusList, List<String> tags) {
        Criteria criteria = new Criteria();
        if (CollectionUtils.isNotEmpty(statusList)) {
            criteria.and("status").in(statusList);
        }
        if (CollectionUtils.isNotEmpty(tags)) {
            criteria.and("tag").in(tags);
        }
        return mqMessageRepository.mqMessageList(Query.query(criteria).skip(skipCount).limit(limit));
    }

    @Override
    public boolean retryRocketMqMessage(String mqMessageId, MqMessage message) {
        if (Objects.isNull(message)) {
            message = mqMessageRepository.selectById(mqMessageId);
        }
        CustomAssert.notNull(message, "消息为空");
        Criteria criteria = new Criteria();
        criteria.and("_id").is(message.getId());
        criteria.and("status").in(RocketMqConstant.FAIL, RocketMqConstant.SENDING);
        Update update = new Update();
        update.inc("version", CommonConstants.ONE);
        update.set("status", RocketMqConstant.RETRY_CANCEL);
        update.set("update_time", LocalDateTime.now());
        update.set("update_by", OperateAuthUtil.getAdministrator());
        mqMessageRepository.update(Query.query(criteria), update);
        //CustomAssert.isTrue(false,"RocketMq消息重试发送失败");
        String result = supplyChainMqProducer.retryMessage(message);
        CustomAssert.isTrue(Objects.equals(result, SendStatus.SEND_OK.name()), "RocketMq消息重试发送失败");
        return Objects.equals(result, SendStatus.SEND_OK.name());
    }

    @Override
    public Page<ChannelMessage> channelMessagePage(ChannelMessageDTO channelMessage) {
		Date startTime = Optional.ofNullable(channelMessage.getStartTime()).map(item -> DateUtil.date(channelMessage.getStartTime())).orElse(null);
		Date endTime = Optional.ofNullable(channelMessage.getEndTime()).map(item -> DateUtil.date(item.plusDays(CommonConstants.ONE))).orElse(null);
        // 创建条件对象
        Criteria criteria = new Criteria();
        criteria.and("del_flag").is(CommonConstants.ZERO);
        if(StringUtils.isNotBlank(channelMessage.getRequestId())) {
            criteria.and("request_id").is(channelMessage.getRequestId());
        }
        if (StringUtils.isNotBlank(channelMessage.getChannelCode())) {
            criteria.and("channel_code").is(channelMessage.getChannelCode());
        }
        if (StringUtils.isNotBlank(channelMessage.getConfigUniqueKey())) {
            criteria.and("config_unique_key").regex(Pattern.compile("^.*" + channelMessage.getConfigUniqueKey() + ".*$", Pattern.CASE_INSENSITIVE));
        }
        if (StringUtils.isNotBlank(channelMessage.getUniqueKey())) {
            criteria.and("unique_key").is(channelMessage.getUniqueKey());
        }
        if (Objects.nonNull(channelMessage.getMessageType())) {
            criteria.and("message_type").is(channelMessage.getMessageType());
        }
        if (StringUtils.isNotBlank(channelMessage.getOperationalCode())) {
            criteria.orOperator(
                    Criteria.where("operational_code").regex(Pattern.compile("^.*" + channelMessage.getOperationalCode() + ".*$", Pattern.CASE_INSENSITIVE)),
                    Criteria.where("channel_operational_code").regex(Pattern.compile("^.*" + channelMessage.getOperationalCode() + ".*$", Pattern.CASE_INSENSITIVE))
            );
        }
        if (Objects.nonNull(channelMessage.getMessageStatus())) {
            criteria.and("status").is(channelMessage.getMessageStatus());
        }
        if (Objects.nonNull(startTime) && Objects.nonNull(endTime)) {
            criteria.andOperator(Criteria.where("create_time").gte(startTime), Criteria.where("create_time").lte(endTime));
        }
        if (Objects.nonNull(startTime) && Objects.isNull(endTime)) {
            criteria.and("create_time").gte(startTime);
        }
        if (Objects.isNull(startTime) && Objects.nonNull(endTime)) {
            criteria.and("create_time").lte(endTime);
        }
        Page<ChannelMessage> page = channelMessageRepository.page(ParamConvert.getPage(channelMessage), Query.query(criteria));
        if (CollectionUtils.isNotEmpty(page.getRecords())) {
            for (ChannelMessage record : page.getRecords()) {
                if (StringUtils.isNotBlank(record.getOriginalMessage())) {
                    record.setOriginalMessage(DeflaterUtils.unzipString(record.getOriginalMessage()));
                }
                if (StringUtils.isNotBlank(record.getMessage())) {
                    record.setMessage(DeflaterUtils.unzipString(record.getMessage()));
                }
            }
        }
        return page;
    }

    @Override
    public Boolean updateChannelMessageStatus(ChannelMessageStatusDTO channelMessageStatus) {
        Criteria criteria = new Criteria();
        criteria.and("_id").is(channelMessageStatus.getChannelMessageId());
        Update update = new Update();
        update.inc("version", CommonConstants.ONE);
        update.set("status", channelMessageStatus.getMessageStatus());
        update.set("remark", channelMessageStatus.getRemark());
        update.set("update_time", LocalDateTime.now());
        update.set("update_by", OperateAuthUtil.getAdministrator());
        return channelMessageRepository.updateChannelMessageStatus(Query.query(criteria), update);
    }

    @Override
    public Boolean channelMessageHandler(String channelMessageId) {
        String tableNameSuffix = DynamicTableNameHelper.getRequestData(SupplyMapperConstant.CHANNEL_MESSAGE_TABLE_NAME);
        Class<? extends ChannelMessage> clazz = Objects.equals(tableNameSuffix, SupplyMapperConstant.CHANNEL_MESSAGE_ORDER_SUFFIX) ? ChannelMessageOrder.class : ChannelMessageProduct.class;
		//手动执行重试需要手动修改重试次数,消费端超过2次重试无效
		//channelMessageRepository.resetRetryCount(channelMessageId,clazz);
        if(Objects.equals(tableNameSuffix, SupplyMapperConstant.CHANNEL_MESSAGE_PRODUCT_SUFFIX)){
            supplyChainMqProducer.sendChannelMessageRetryMessage(channelMessageId);
            return Boolean.TRUE;
        } else {
            ChannelMessage channelMessage = channelMessageRepository.getById(channelMessageId, clazz);
            CustomAssert.notNull(channelMessage, "消息记录不存在");
            if (EnumUtils.eq(MessageStatusEnum.SUCCESS, channelMessage.getStatus())) {
                if(log.isDebugEnabled()) {
                    log.debug("渠道消息异步处理器 消息状态处理成功: message : {}", JSON.toJSONString(channelMessage));
                }
                return true;
            }
            ChannelMessageBO bo = channelMessageConverter.toChannelMessage(channelMessage);
            AbstractChannelMessageAsyncHandler handler = FrameworkUtils.strategyInstance(AbstractChannelMessageAsyncHandler.class, bo.getChannelCode(), bo.getMessageType());
            CustomAssert.notNull(handler, String.format("渠道消息异步处理器[AbstractChannelMessageAsyncHandler]不存在(channelCode:%s , messageType:%s)", bo.getChannelCode(), bo.getMessageType()));
            return handler.strategyHandler(bo);
        }
    }

    @Override
    public Boolean updateDelayMessageStatus(DelayMessageStatusDTO delayMessageStatus) {
        Criteria criteria = new Criteria();
        criteria.and("_id").is(delayMessageStatus.getDelayMessageId());
        Update update = new Update();
        update.inc("version", CommonConstants.ONE);
        update.set("status", delayMessageStatus.getMessageStatus());
        update.set("remark", delayMessageStatus.getRemark());
        update.set("update_time", LocalDateTime.now());
        update.set("update_by", OperateAuthUtil.getAdministrator());
        return delayMessageRepository.updateDelayMessageStatus(Query.query(criteria), update);
    }

    @Override
    public Boolean delayMessageHandler(String delayMessageId) {
        DelayMessage delayMessage = delayMessageRepository.findById(delayMessageId);
		distributionQueueJobConsumerWork.consume(delayMessage);
        return Boolean.TRUE;
    }

    @Override
    public Page<DelayMessage> delayMessagePage(DelayMessageDTO delayMessage) {
		Date startTime = Optional.ofNullable(delayMessage.getStartTime()).map(item -> DateUtil.date(delayMessage.getStartTime())).orElse(null);
		Date endTime = Optional.ofNullable(delayMessage.getEndTime()).map(item -> DateUtil.date(item.plusDays(CommonConstants.ONE))).orElse(null);
        // 创建条件对象
        Criteria criteria = new Criteria();
        criteria.and("del_flag").is(CommonConstants.ZERO);
        if(StringUtils.isNotBlank(delayMessage.getUniqueKey())) {
            criteria.and("unique_key").is(delayMessage.getUniqueKey());
        }
        if (Objects.nonNull(delayMessage.getMessageType())) {
            criteria.and("message_type").is(delayMessage.getMessageType());
        }
        if (Objects.nonNull(delayMessage.getMessageStatus())) {
            criteria.and("status").is(delayMessage.getMessageStatus());
        }
        if (StringUtils.isNotBlank(delayMessage.getOperationalCode())) {
            criteria.and("operational_code").regex(Pattern.compile("^.*" + delayMessage.getOperationalCode() + ".*$", Pattern.CASE_INSENSITIVE));
        }
        if (Objects.nonNull(startTime) && Objects.nonNull(endTime)) {
            criteria.andOperator(Criteria.where("create_time").gte(startTime), Criteria.where("create_time").lte(endTime));
        }
        if (Objects.nonNull(startTime) && Objects.isNull(endTime)) {
            criteria.and("create_time").gte(startTime);
        }
        if (Objects.isNull(startTime) && Objects.nonNull(endTime)) {
            criteria.and("create_time").lte(endTime);
        }
        return delayMessageRepository.page(ParamConvert.getPage(delayMessage), Query.query(criteria));
    }
}
